package com.example.network;

import com.alibaba.fastjson.JSON;
import com.example.model.AllNodeInfo;
import com.example.model.Message;
import com.example.model.NodeInfo;
import com.example.util.CommonUtil;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

@Component
public class Server {
    private final static int size = 1024;
    private final static int server_port = 7150;

    public void start() {
        Thread thread = new Thread(new connection());
        thread.start();
    }

    private class connection implements Runnable {
        @Override
        public void run() {
            try (Selector selector = Selector.open();
                    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {

                serverSocketChannel.socket().bind(new InetSocketAddress(server_port));
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                String localIp = CommonUtil.getLocalIp();
                System.out.println("锚节点启动 " + localIp + ":" + server_port);

                while (true) {
                    selector.select();
                    Set<SelectionKey> selectedKeys = selector.selectedKeys();
                    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                    while (keyIterator.hasNext()) {
                        SelectionKey key = keyIterator.next();

                        if (key.isAcceptable()) {
                            handleAccept(key, selector);
                        } else if (key.isReadable()) {
                            handleRead(key);
                        }
                        keyIterator.remove();
                    }
                }
            } catch (IOException e) {
                System.out.println("监听异常" + e.getMessage());
            }
        }
    }

    private void handleAccept(SelectionKey key, Selector selector) throws IOException {
        ServerSocketChannel serverchannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverchannel.accept();
        clientChannel.configureBlocking(false);
        clientChannel.register(selector, SelectionKey.OP_READ);
    }

    private void handleRead(SelectionKey key) {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        try {

            ByteBuffer buffer = ByteBuffer.allocate(Server.size);
            int bytesRead = socketChannel.read(buffer);

            if (bytesRead > 0) {
                buffer.flip();
                byte[] data = new byte[buffer.remaining()];
                buffer.get(data);
                String message = new String(data);
                System.out.println("接收 " + socketChannel.getRemoteAddress() + ": " + message);
                handleMessage(message, socketChannel);
            } else if (bytesRead == -1) {
                System.out.println("节点: " + socketChannel.getRemoteAddress() + "断开连接");
                socketChannel.close();
            }
        } catch (Exception e) {
            AllNodeInfo.channels.clear();
            AllNodeInfo.nodes.clear();
            close(socketChannel);
        }
    }

    private void handleMessage(String msg, SocketChannel channel) {
        Message message = JSON.parseObject(msg, Message.class);
        System.out.println("处理消息来自: " + channel.socket().getRemoteSocketAddress() +
                ": " + JSON.toJSONString(message));
        String data = message.getData();
        String[] parts = data.split(":");
        NodeInfo node = new NodeInfo();
        node.setIP(parts[0]);// ip
        node.setPort(Integer.parseInt(parts[1]));// port
        AllNodeInfo.setNodes(message.getNodeIndex(), node);// 保存节点
        AllNodeInfo.setChannels(message.getNodeIndex(), channel);// 保存通道
        AllNodeInfo.printNodes();
        Message response = new Message();
        response.setType(-1);
        response.setData(JSON.toJSONString(AllNodeInfo.nodes));
        response.setNodeIndex("Anchor");
        broadcast(response);
    }

    // 发送消息到指定连接
    private boolean sendMessage(SocketChannel channel, String msg) {
        try {
            ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
            System.out.println("发送消息:" + msg + "-->" + channel.socket().getRemoteSocketAddress());
            channel.write(buffer);
            return true;
        } catch (IOException e) {
            System.err.println("发送消息失败: " + e.getMessage());
            close(channel);
            return false;
        }
    }

    // 关闭指定连接
    private void close(SocketChannel channel) {
        try {
            channel.close();
        } catch (Exception e) {
            System.out.println("关闭指定连接 " + e.getMessage());
        }
    }

    /**
     * 向所有连接的节点广播消息
     */
    public void broadcast(Message message) {
        if (AllNodeInfo.channels.isEmpty()) {
            System.out.println("没有连接任何节点");
            return;
        }

        String msg = JSON.toJSONString(message);
        System.out.println("======全网广播消息开始======");

        // 广播
        AllNodeInfo.channels.forEach((index, channel) -> {
            if (channel.isConnected()) {
                sendMessage(channel, msg);
            }
        });

        System.out.println("======全网广播消息结束======");
    }
}